package iterator;

import global.AttrType;
import global.RID;
import global.SystemDefs;
import heap.FieldNumberOutOfBoundException;
import heap.Heapfile;
import heap.InvalidTupleSizeException;
import heap.InvalidTypeException;
import heap.Scan;
import heap.Tuple;
import index.IndexException;

import java.io.IOException;

import columnar.BitmapIndexScan;
import columnar.Columnarfile;

import bufmgr.PageNotReadException;

public class ColumnarBitmapEquiJoins  extends Iterator 
{
	private AttrType      _in1[],  _in2[];
	private   int        in1_len, in2_len;

	private   short t2_str_sizescopy[];
	private   CondExpr OutputFilter[];
	private   CondExpr RightFilter[];
	private   int        n_buf_pgs;        // # of buffer pages available.
	private   boolean        done,         // Is the join complete
	get_from_outer;                 // if TRUE, a tuple is got from outer
	private   Tuple     outer_tuple, inner_tuple;
	private   Tuple     Jtuple;           // Joined tuple
	private   FldSpec   perm_mat[];
	private   int        nOutFlds;
	private int leftIndexFileCursor;
  //private   Heapfile  hf;
  //private   Scan      inner;
	private Columnarfile lcf = null;
	private Columnarfile rcf = null;
	private BitmapIndexScan bmOuterScan = null;
	private BitmapIndexScan bmInnerScan = null;
	private boolean get_new_indexfile;
	private String[] indName;
	
    Tuple outerIndPos = null;
    Tuple innerIndPos = null;
    int outPosition;
    int innerPosition;
	  
	  /**constructor
	   *Initialize the two relations which are joined, including relation type,
	   *@param in1  Array containing field types of R.
	   *@param len_in1  # of columns in R.
	   *@param t1_str_sizes shows the length of the string fields.
	   *@param in2  Array containing field types of S
	   *@param len_in2  # of columns in S
	   *@param  t2_str_sizes shows the length of the string fields.
	   *@param amt_of_mem  IN PAGES
	   *@param am1  access method for left i/p to join
	   *@param relationName  access hfapfile for right i/p to join
	   *@param outFilter   select expressions
	   *@param rightFilter reference to filter applied on right i/p
	   *@param proj_list shows what input fields go where in the output tuple
	   *@param n_out_flds number of outer relation fileds
	   *@exception IOException some I/O fault
	   *@exception NestedLoopException exception from this class
	   */
	public ColumnarBitmapEquiJoins( AttrType    in1[],    
				   int     len_in1,           
				   short   t1_str_sizes[],
				   AttrType    in2[],         
				   int     len_in2,           
				   short   t2_str_sizes[],   
				   int     amt_of_mem,  
				   String leftColumnarFileName,
				   int leftJoinField,
				   String rightColumnarFileName,
				   int rightJoinField,   
				   FldSpec   proj_list[],
				   int        n_out_flds
				   ) throws IOException,NestedLoopException
	    {
		
	      
	      try {
			lcf = new Columnarfile(leftColumnarFileName);
			rcf = new Columnarfile(rightColumnarFileName);
	      } catch (Exception ex) {
			System.err.println("Unable to read a columnar files from DB\n\t check if they have been created.\n");
			Runtime.getRuntime().exit(1);
	      }

		
	      _in1 = new AttrType[in1.length];
	      _in2 = new AttrType[in2.length];
	      System.arraycopy(in1,0,_in1,0,in1.length);
	      System.arraycopy(in2,0,_in2,0,in2.length);
	      in1_len = len_in1;
	      in2_len = len_in2;
	      
	      t2_str_sizescopy =  t2_str_sizes;
	      inner_tuple = new Tuple();
	      Jtuple = new Tuple();
	      n_buf_pgs    = amt_of_mem;
	      done  = false;
	      get_new_indexfile = true;
	      get_from_outer = true;
	      
	      indName = null;
	      
	      AttrType[] Jtypes = new AttrType[n_out_flds];
	      short[]    t_size;
	      
	      perm_mat = proj_list;
	      nOutFlds = n_out_flds;
	      try {
	    	  t_size = TupleUtils.setup_op_tuple(Jtuple, Jtypes,
							   in1, len_in1, in2, len_in2,
							   t1_str_sizes, t2_str_sizes,
							   proj_list, nOutFlds);
	      }catch (TupleUtilsException e)
	      {
	    	  throw new NestedLoopException(e,"TupleUtilsException is caught by NestedLoopsJoins.java");
	      }
	      
	      leftIndexFileCursor = 0;
	      outPosition =  0;
	      innerPosition = 0;


	    }
	  
  /**  
   *@return The joined tuple is returned
   *@exception IOException I/O errors
   *@exception JoinsException some join exception
   *@exception IndexException exception from super class
   *@exception InvalidTupleSizeException invalid tuple size
   *@exception InvalidTypeException tuple type not valid
   *@exception PageNotReadException exception from lower layer
   *@exception TupleUtilsException exception from using tuple utilities
   *@exception PredEvalException exception from PredEval class
   *@exception SortException sort exception
   *@exception LowMemException memory error
   *@exception UnknowAttrType attribute type unknown
   *@exception UnknownKeyTypeException key type unknown
   *@exception Exception other exceptions

   */
	public Tuple get_next()
	    throws IOException,
		   JoinsException ,
		   IndexException,
		   InvalidTupleSizeException,
		   InvalidTypeException, 
		   PageNotReadException,
		   TupleUtilsException, 
		   PredEvalException,
		   SortException,
		   LowMemException,
		   UnknowAttrType,
		   UnknownKeyTypeException,
		   Exception
	{

		if (done)
			return null;
		
		do
		{
			if (get_from_outer)
			{
				get_from_outer = false;
				
				if(get_new_indexfile)
				{
					get_new_indexfile = false;
					indName = getNextJointIndexName();
					
					if(indName==null)
					{
						done = true;
						return null;
					}
					
					if (bmOuterScan != null)     // If this not the first time,
					{
						if(!bmOuterScan.closeFlag)
							bmOuterScan.close();
						bmOuterScan = null;
					}
					
					try {
						bmOuterScan = new BitmapIndexScan(indName[0]);
			        }
			        catch (Exception e) {
			        	e.printStackTrace();
			        }
				
				}
				

				if (bmInnerScan != null)     // If this not the first time,
				{
					if(!bmInnerScan.closeFlag)
						bmInnerScan.close();
					bmInnerScan = null;
				}
				
				try {
					 bmInnerScan = new BitmapIndexScan(indName[1]);
				}
				catch(Exception e){
					throw new NestedLoopException(e, "openIndexScan failed");
				}

				if ((outerIndPos=bmOuterScan.get_next()) == null)
				{
					
					get_new_indexfile = true;
					get_from_outer = true;
					
					if (bmInnerScan != null) 
					{
						if(!bmInnerScan.closeFlag)
							bmInnerScan.close();
						bmInnerScan = null;
					}
					  
					continue;//return null;
				}   
	           outPosition = outerIndPos.getIntFld(1);
	           outer_tuple = lcf.getCompleteTupleByPosition(outPosition);
	        
			}

			while ((innerIndPos = bmInnerScan.get_next()) != null)
			{
				int innerPosition = innerIndPos.getIntFld(1);
				inner_tuple = rcf.getCompleteTupleByPosition(innerPosition);
				
				if (PredEval.Eval(RightFilter, inner_tuple, null, _in2, null) == true)
				{
					if (PredEval.Eval(OutputFilter, outer_tuple, inner_tuple, _in1, _in2) == true)
					{
						// Apply a projection on the outer and inner tuples.
						Projection.Join(outer_tuple, _in1, 
									  inner_tuple, _in2, 
									  Jtuple, perm_mat, nOutFlds);
						return Jtuple;
					}
				}  
			}

			get_from_outer = true; // Loop back to top and get next outer tuple.	
			
		} while (true);

	} 
	 
  /**
   * implement the abstract method close() from super class Iterator
   *to finish cleaning up
   *@exception IOException I/O error from lower layers
   *@exception JoinsException join error from lower layers
   *@exception IndexException index access error 
   */
	public void close() throws JoinsException, IOException,IndexException 
	{
		if (!closeFlag) 
		{
			try {
		//		if(!bmOuterScan.closeFlag)
		//			bmOuterScan.close();
	        
			}catch (Exception e) {
				throw new JoinsException(e, "ColumnarBitmapEquiJoins.java: error in closing bitmap scans.");
			}
			closeFlag = true;
		}
	}
	
	private String[] getNextJointIndexName() throws Exception
	{
		while(leftIndexFileCursor < lcf.indexFileList().size())
		{
			String indNameL = lcf.indexFileList().get(leftIndexFileCursor++).toString();
			String indNameR = indNameL.replaceFirst(lcf.columnarFileName(), rcf.columnarFileName());
			
			if(rcf.indexFileList().contains(indNameR))
				return new String[]{indNameL, indNameR};
		}
		return null;
	}
}
